package com.ikas.ai.framework;

import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.ikas.ai.framework.factory.ScoreThreadFactory;
import com.ikas.ai.framework.flow.ScoreFlow;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @Description: 评分流程执行EventLoop
 * @Author: qi.yilai
 * @Date: 2022/8/11 19:10
 * @Version: 1.0.0
 **/
@Slf4j
public class FlowEventLoop {
    private static final int READY_THRESHOLD_SIZE = 1024;
    private static final int DEFAULT_CORE_SIZE = Runtime.getRuntime().availableProcessors() * 2;
    private static final int DEFAULT_POOL_SIZE = 100;
    private static final int DEFAULT_KEEP_ALIVE = 30;
    private static final int DEFAULT_BLOCKING_CAPACITY = 1024;
    private static final int DEFAULT_CHECK_INTERVAL = 30;
    private static final int DEFAULT_FLOW_CAPACITY_THRESHOLD = 10000;

    private int checkInterval;

    // 阻塞队列缓冲区
    private ArrayBlockingQueue<ScoreFlow> readyQueues;
    // 执行调用flow线程池
    private ExecutorService executorService;
    // flow pool缓存池（流程实例 -> 流程）
    private Map<Long, ScoreFlow> flowPool = Maps.newConcurrentMap();
    // flow pool缓存池（流程id -> 流程）
    private Map<Long, ScoreFlow> idFlowPool = Maps.newConcurrentMap();
    // 定时检查flow状态
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
    // 挂起暂停的流程
    private Map<Long, ScoreFlow> pendingFlows = Maps.newConcurrentMap();
    // 流程引擎限制最大流程数量
    private int flowThreshold;

    public FlowEventLoop(int coreSize, int executorPoolSize, int checkInterval, int readyThresholdSize, int flowThreshold) {
        this.checkInterval = checkInterval;
        this.readyQueues = Queues.newArrayBlockingQueue(readyThresholdSize);
        this.executorService = new ThreadPoolExecutor(coreSize, executorPoolSize, DEFAULT_KEEP_ALIVE, TimeUnit.SECONDS, new ArrayBlockingQueue(DEFAULT_BLOCKING_CAPACITY), new ScoreThreadFactory());
        this.flowThreshold = flowThreshold;
    }

    public FlowEventLoop() {
        this(DEFAULT_CORE_SIZE, DEFAULT_POOL_SIZE, DEFAULT_CHECK_INTERVAL, READY_THRESHOLD_SIZE, DEFAULT_FLOW_CAPACITY_THRESHOLD);
    }

    public void addFlow(ScoreFlow flow) {
        if (size() > flowThreshold) {
            log.warn("流程引擎数量超过{}, 当前最大阈值数量{}，尽可能进行扩容处理", size(), flowThreshold);
        }
        try {
            readyQueues.put(flow);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("addFlow error", e);
        }
    }

    public void start() {
        // 检测流程是否执行
        startCheckFlow();
        // 执行流程
        startExecuteFlow();
    }

    /**
     * 停止正在执行的流程
     *
     * @param flowInstanceId
     */
    public boolean stopFlow(long flowInstanceId) {
        ScoreFlow doingFlow = flowPool.get(flowInstanceId);
        if (Objects.nonNull(doingFlow)) {
            boolean flag = removeFlow(doingFlow);
            doingFlow.stop();
            log.info("停止流程{}成功", doingFlow.getFlowInstanceId());
            return flag;
        }
        return false;
    }

    public boolean removeFlow(ScoreFlow flow) {
        idFlowPool.remove(flow.getFlowId());
        return flowPool.remove(flow.getFlowInstanceId()) != null;
    }

    /**
     * 判断是否正在执行
     *
     * @param flowInstanceId
     * @return
     */
    public boolean isExecuting(long flowInstanceId) {
        return flowPool.containsKey(flowInstanceId);
    }

    /**
     * 判断是否正在执行
     *
     * @param flowId
     * @return
     */
    public boolean isExecutingForId(long flowId) {
        return idFlowPool.containsKey(flowId);
    }

    private void startExecuteFlow() {
        executorService.submit(() -> {
            while (true) {
                ScoreFlow flow = takeFlow();
                if (Objects.isNull(flow)) {
                    sleep();
                    continue;
                }
                submitFlow(flow);
            }
        });
    }


    private void sleep() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.error("sleep error", e);
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 检查 flow
     */
    private void startCheckFlow() {
        log.info("执行检查流程状态定时任务");
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                List<ScoreFlow> flows = new ArrayList<>(flowPool.values());
                for (ScoreFlow flow : flows) {
                    if (flow.isOver()) {
                        log.info("检测到流程实例【{}】已经完成或者异常无法继续执行，从流程池进行清除", flow.getFlowInstanceId());
                        removeFlow(flow);
                    }
                }
            } catch (Exception exception) {
                log.error("检测流程出现异常", exception);
            }
            int executingCount = flowPool.size();
            int blockCount = readyQueues.size();
//            log.info("当前流程处理器正在执行流程数量：{}， 阻塞的流程数量：{}", executingCount, blockCount);
        }, checkInterval, checkInterval, TimeUnit.SECONDS);
    }


    private boolean submitFlow(ScoreFlow flow) {
        flowPool.put(flow.getFlowInstanceId(), flow);
        idFlowPool.put(flow.getFlowId(), flow);
        // 全流程自动执行
        return flow.start(executorService);
    }

    private ScoreFlow takeFlow() {
        ScoreFlow flow = null;
        try {
            flow = readyQueues.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("take flow error", e);
        }
        return flow;
    }

    public int size() {
        return flowPool.size() + readyQueues.size();
    }
}
